package com.doit.day04

import scala.io.Source

object _10_电影案例 {
  def main(args: Array[String]): Unit = {
    //读取数据
    val movies: List[String] = Source.fromFile("C:\\Users\\34650\\Desktop\\课程资料\\资料\\movies.txt").getLines().toList
    val users: List[String] = Source.fromFile("C:\\Users\\34650\\Desktop\\课程资料\\资料\\users.txt").getLines().toList
    val ratings: List[String] = Source.fromFile("C:\\Users\\34650\\Desktop\\课程资料\\资料\\ratings.txt").getLines().toList
    //    xuqiu1(movies,ratings)

    //    xuqiu2(movies,ratings)

    //    xuqiu3(movies, ratings)

    //    xuqiu4(users,ratings)

    val start: Long = System.currentTimeMillis()
    //    xuqiu4(users,ratings)
//    xuqiu4_liang(users, ratings)
    xuqiu5(users, ratings)
    val end: Long = System.currentTimeMillis()
    println(end - start)
  }

  def movieIdGetInfo(movies: List[String]) = {
    val movieMap: Map[String, (String, String)] = movies.map(line => {
      val arr: Array[String] = line.split("_")
      //电影id   (电影名称，电影类型)
      (arr(0), (arr(1), arr(2)))
    }).toMap
    movieMap
  }

  def etlRatings(ratings: List[String]) = {
    val nameAndRatings = ratings.map(line => {
      val arr: Array[String] = line.split("_")
      //1_1193_5_978300760
      (arr(0), arr(1), arr(2), arr(3))
    }
    )
    nameAndRatings
  }

  //1.求每部电影的平均分，按照降序输出  ==> 我需要得是电影名称 + 平均分
  def xuqiu1(movies: List[String], ratings: List[String]) = {
    val movieMap: Map[String, (String, String)] = movieIdGetInfo(movies)

    val nameAndRatings: List[(String, String)] = ratings.map(line => {
      val arr: Array[String] = line.split("_")
      //1_1193_5_978300760
      val movie_id: String = arr(1)
      val nameAndType: (String, String) = movieMap.getOrElse(movie_id, ("找不着", "找不着"))
      //要求什么？？？每一部电影名称的平均分
      (nameAndType._1, arr(2))
    })

    //String  电影名称  list里面放的是相同的电影名称和分数
    val map: Map[String, List[(String, String)]] = nameAndRatings.groupBy(_._1)

    map.map(tp => {
      val value: List[(String, String)] = tp._2
      val listDoubles: List[Double] = value.map(_._2.toDouble)
      (tp._1, (listDoubles.sum / listDoubles.size).formatted("%.2f").toDouble)
    }).toList
      .sortBy(_._2)
      .reverse
      .take(20)
      .foreach(println)
  }

  //2.每个用户最早评价的前三部电影(用户，电影信息，评分值)
  def xuqiu2(movies: List[String], ratings: List[String]) = {
    val movieMap: Map[String, (String, String)] = movieIdGetInfo(movies)
    val etlRat: List[(String, String, String, String)] = etlRatings(ratings)
    //先分组  按照用户id分组
    val map: Map[String, List[(String, String, String, String)]] = etlRat.groupBy(_._1)
    //按照时间戳进行排序，然后取前3
    map.toList.map(tp => {
      val user_id: String = tp._1
      val top3: List[(String, String, String, String)] = tp._2.sortBy(_._4.toLong).take(3)
      val tuples: List[(String, String, String, String, String)] = top3.map(tp => {
        val movie_id: String = tp._2
        //根据电影id获取到电影名称和电影类型
        val tuple: (String, String) = movieMap.getOrElse(movie_id, ("找不着", "找不着"))
        (tp._2, tuple._1, tuple._2, tp._3, tp._4)
      })
      (user_id, tuples)
    }).foreach(println)
  }

  //3.哪些年份的电影评分(平均分)最高，取最高的前五个年份
  //Antonia's Line (Antonia) (1995)
  def xuqiu3(movies: List[String], ratings: List[String]): Unit = {
    //获取movie信息的时候，要的是电影id和年份
    val map: Map[String, (String, String)] = movieIdGetInfo(movies)

    //获取ratings的时候，要的是电影id和评分
    val tuples: List[(String, String, String, String)] = etlRatings(ratings)

    val yearAndRatings: List[(String, Double)] = tuples.map(tp => {
      val nameAndType: (String, String) = map.getOrElse(tp._2, ("not found", "not found"))
      val year: String = nameAndType._1.substring(nameAndType._1.length - 5, nameAndType._1.length - 1)
      (year, tp._3.toDouble)
    })

    yearAndRatings.groupBy(_._1)
      .map(tp => (tp._1, (tp._2.map(_._2).sum / tp._2.size).formatted("%.2f").toDouble))
      .toList
      .sortBy(_._2)
      .reverse
      .take(5)
      .foreach(println)
  }

  //  4.每个职业最喜欢的前三个电影id   观看次数最多的
  def xuqiu4(users: List[String], ratings: List[String]) = {
    //把每个user处理一下
    val userIDAndJob: Map[String, String] = users.map(line => {
      //1_F_1_10_48067
      val arr: Array[String] = line.split("_")
      (arr(0), arr(3))
    }).toMap

    //处理评分表  用户id    电影id    评分   时间戳
    val tuples: List[(String, String, String, String)] = etlRatings(ratings)

    //和评分进行关联    100万次
    val jobAndMovieId: List[(String, String)] = tuples.map(tp => {
      val job: String = userIDAndJob.getOrElse(tp._1, "未知")
      (job, tp._2)
    })


    val jobAndIds: Map[String, List[(String, String)]] = jobAndMovieId.groupBy(_._1)

    jobAndIds.map(tp => {
      val job: String = tp._1
      val movieIds: List[String] = tp._2.map(_._2)
      val idAndCnt: List[(String, Int)] = movieIds.groupBy(word => word)
        .map(tp => (tp._1, tp._2.size))
        .toList
        .sortBy(_._2)
        .reverse
        .take(3)
      (job, idAndCnt)
    }).foreach(println)


  }

  def xuqiu4_liang(users: List[String], ratings: List[String]) = {
    //把每个user处理一下
    val userIDAndJob: Map[String, String] = users.map(line => {
      //1_F_1_10_48067
      val arr: Array[String] = line.split("_")
      (arr(0), arr(3))
    }).toMap

    //处理评分表  用户id    电影id    评分   时间戳
    val tuples: List[(String, String, String, String)] = etlRatings(ratings)

    //上来就分组  用户id
    val map: Map[String, List[(String, String, String, String)]] = tuples.groupBy(_._1)

    val map1: List[(String, List[String])] = map.toList.map(tp => {
      val user_id: String = tp._1
      val job: String = userIDAndJob.getOrElse(user_id, "未知")
      val value: List[(String, String, String, String)] = tp._2
      val movieIds = value.map(_._2)
      (job, movieIds)
    })

    val grouped: Map[String, List[(String, List[String])]] = map1.groupBy(_._1)
    grouped.map(tp => {
      val job: String = tp._1

      val list: List[String] = tp._2.map(_._2).flatten

      val tuples1: List[(String, Int)] = list.groupBy(word => word)
        .map(tp => (tp._1, tp._2.size))
        .toList
        .sortBy(_._2)
        .reverse
        .take(3)
      (job, tuples1)
    }).foreach(println)
  }


  //  5.年龄段在“18-24”的男性年轻人，最喜欢看哪10部电影（Id） 平均分 最多的10个电影
  def xuqiu5(users: List[String], ratings: List[String]) = {
    val userIDAndJob: List[String] = users.map(line => {
      //1_F_1_10_48067
      val arr: Array[String] = line.split("_")
      //arr(1) => M  arr(2) = 18
      (arr(0), (arr(1), arr(2)))
    }).filter(tp=>{
      tp._2._1.equals("M") && tp._2._2.equals("18")
    }).map(_._1)

    //处理评分表  用户id    电影id    评分   时间戳
    val tuples: List[(String, String, String, String)] = etlRatings(ratings)

    val map: Map[String, List[(String, String, String, String)]] = tuples.groupBy(_._1)

    val userIdAndAvg: Map[String, List[(String, Double)]] = map.map(tp => {
      val user_id: String = tp._1
      //每个用户的平均分   每个用户看电影的平均分
      val list: List[(String, String, String, String)] = tp._2
      //针对电影id进行分组，然后获取到每一个用户(大的逻辑，已经分好了)
      val idAndRatings: Map[String, List[(String, String, String, String)]] = list.groupBy(_._2)

      val idAndRatings1: List[(String, Double)] = idAndRatings.map(tp => {
        val doubles: List[Double] = tp._2.map(_._3.toDouble)
        (tp._1, (doubles.sum / doubles.size).formatted("%.2f").toDouble)
      }).toList
        .sortBy(_._2)
        .reverse
        .take(10)
      (user_id,idAndRatings1 )
    })

    val userAndAVG: List[(String, List[(String, Double)])] = userIDAndJob.map(user_id => {
      val tuples1: List[(String, Double)] = userIdAndAvg.getOrElse(user_id, List(("未知", 0.0)))
      (user_id, tuples1)
    })
    userAndAVG.foreach(println)
  }

}
